package com.test.writer
import java.sql.{Connection, DriverManager, Statement}

import com.test.bean.BusInfo
import org.apache.spark.sql.ForeachWriter

class MysqlWriter extends ForeachWriter[BusInfo] {

  var connection: Connection = _

  //初始化连接
  override def open(partitionId: Long, epochId: Long): Boolean = {
    Class.forName("com.mysql.jdbc.Driver")
    val url =
      "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=false"
    val user = "root"
    val pwd = "root"
    connection = DriverManager.getConnection(url, user, pwd)
    true
  }

  //处理数据
  override def process(value: BusInfo): Unit = {
    //rowkey:调度编号+车牌号+时间戳
    var rowkey = value.deployNum + value.plateNum + value.timeStr
    val arr: Array[String] = value.lglat.split("_")

    val statement: Statement = connection.createStatement()
    val bool: Boolean = statement.execute("insert into bus_info(id,lng,lat) values ('"+rowkey+"','"+arr(0)+"','"+arr(1)+"')")


  }

  //关闭连接
  override def close(errorOrNull: Throwable): Unit = {

    if (connection != null) {
      connection.close()
    }

  }
}
